<?php

namespace app\rabbitmq\controller;

use PhpAmqpLib\Connection\AMQPStreamConnection;
//引入rabbitMq所需类
use PhpAmqpLib\Message\AMQPMessage;
use think\Config;

class MqProducer {
    /**
     * 生产者
     * @param $data 控制器传来的数据
     * @param $name 配置文件下rabbitmq文件前缀名
     * @throws \Exception
     */
    public function publish($data, $name, $multiple) {
        //获取rabbitMq配置(刚刚在config文件下配置的rabbitmq信息)
        $amqp       = Config::get('AMQP');
        $amqpDefail = Config::get('' . $name . '_queue');

        //建立连接
        $connection = new AMQPStreamConnection(
            $amqp['host'],
            $amqp['port'],
            $amqp['username'],
            $amqp['password']
        );

        //建立通道
        $channel = $connection->channel();

        //初始化交换机
        $channel->exchange_declare($amqpDefail['exchange_name'], $amqpDefail['exchange_type'], false, true, false);

        //初始化队列
        $channel->queue_declare($amqpDefail['queue_name'], false, true, false, false);

        //绑定队列到交换机
        $channel->queue_bind($amqpDefail['queue_name'], $amqpDefail['exchange_name'], $amqpDefail['route_key']);

        if ($multiple) {
            foreach ($data as $value) {
                $msg = new AMQPMessage(json_encode($value, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
                $channel->basic_publish($msg, $amqpDefail['exchange_name'], $amqpDefail['route_key']);
            }
        } else {
            $msg = new AMQPMessage(json_encode($data, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
            $channel->basic_publish($msg, $amqpDefail['exchange_name'], $amqpDefail['route_key']);
        }


        //关闭通道
        $channel->close();

        //断开连接
        $connection->close();
    }
}